#include "pipe_session.h"
#include "pipe_protocol.h"
#include "pipe_module.h"
#include "client.hpp"
#include "log.h"

namespace anet {
	namespace pipe {
		CPipeSession::CPipeSession() : m_lastBeatHeart(GetTickCount()),
			m_remoteId(0), m_token(""), m_session(nullptr), 
			m_mode(ePipeMode::pipe_mode_listener), 
			m_state(ePipeState::pipe_session_state_invalid),
			m_pipeHandler(nullptr), m_client(nullptr), 
			m_timer(CPipeModule::instance().getTimeWheel()) {
		}

		CPipeSession::~CPipeSession() {
			if (m_mode == ePipeMode::pipe_mode_connector) {
				assert(m_client != nullptr && "client pointer is nullptr");
				delete m_client;
				m_client = nullptr;
			}
			m_pipeHandler = nullptr;
			m_session = nullptr;
		}

		void CPipeSession::SetRemoteId(uint32 id) {
			m_remoteId = id;
		}

		void CPipeSession::SetToken(const std::string& token) {
			m_token = token;
		}

		void CPipeSession::SetMode(ePipeMode mode) {
			m_mode = mode;
		}

		void CPipeSession::SetMsgHandler(IPipeMsgHandler* pipeHandler) {
			assert(pipeHandler != nullptr && "msg handler is nullptr");
			if (pipeHandler == nullptr) {
				return;
			}
			m_pipeHandler = pipeHandler;
		}

		uint32 CPipeSession::GetPipeId() const {
			return CPipeModule::instance().GetLocalId();
		}

		SPipeId CPipeSession::GetPipeInfo() const {
			return getPipeId(this->GetPipeId());
		}

		uint32 CPipeSession::GetRemotePipeId() const {
			return m_remoteId;
		}

		SPipeId CPipeSession::GetRemotePipeInfo() const {
			return getPipeId(this->GetRemotePipeId());
		}

		IPipeMsgHandler* CPipeSession::GetMsgHandler() {
			return m_pipeHandler;
		}

		void CPipeSession::Close() {
			m_session->Close();
		}

		void CPipeSession::Send(const char* msg, int len) {
			if (m_state != ePipeState::pipe_session_state_pipe_connected) {
				return;
			}
			if (msg == nullptr) {
				// ignore empty message.
				return;
			}

			// build message.
			std::vector<char> buffer;
			buffer.resize(sizeof(anet::tcp::SCommonHead) + len);

			char* pBuff = &buffer[0];
			anet::tcp::SCommonHead* pHead = (anet::tcp::SCommonHead*)pBuff;
			pHead->len = (decltype(pHead->len))htonl(len);

			memcpy(pBuff + sizeof(anet::tcp::SCommonHead), msg, len);
			m_session->Send(pBuff, int(sizeof(anet::tcp::SCommonHead) + len));
		}

		void CPipeSession::SetClient(anet::tcp::CClient* client) {
			assert(m_mode == ePipeMode::pipe_mode_connector && 
				client != nullptr &&
				"only connector mode can set client"
			);
			m_client = client;
		}

		CPipeSession* CPipeSession::Create() {
			return new CPipeSession();
		}

		void CPipeSession::SetSession(sharePtrSession* pSession) {
			m_session = pSession;
		}

		void CPipeSession::OnMessage(const char* msg, int len) {
			if (len < int(sizeof(anet::tcp::SCommonHead))) {
				return;
			}

			// set beat heart time(all the message are as beat heart message).
			m_lastBeatHeart = GetTickCount();

			switch (m_state) {
			case ePipeState::pipe_session_state_net_connected: {
				if (len < int(sizeof(anet::tcp::SCommonHead) + sizeof(uint16))) {
					LogAdebug("{} find invalid packet", getPipeIdInfo(this->GetPipeId()));
					return ;
				}

				// message id.
				uint16 msgId = ntohs(*(uint16*)(msg + sizeof(anet::tcp::SCommonHead)));

				if (m_mode == ePipeMode::pipe_mode_listener) {
					// listen peer.
					if (msgId == ePipeMessageId::pipe_message_connect_req_id) {
						// connect_req message.
						connect_req* req = (connect_req*)(msg + sizeof(anet::tcp::SCommonHead) + sizeof(uint16));
						if (!CPipeModule::instance().CheckConnectInfo(req->remote_id, req->token)) {
							LogAdebug("remote id:{},token:{} is not the same with local information", getPipeIdInfo(req->remote_id), req->token);
							m_session->Close();
							return;
						}

						m_state = ePipeState::pipe_session_state_pipe_connected;
						m_remoteId = req->local_id;

						// send ack message.
						this->sendConnectAck();

						// call pipe connected reporter.
						CPipeModule::instance().GetReporter()->OnReport(true, this);

						// start timer to check beat heat.
						this->startTimerToCheckBeatHeart();
					} else if (msgId == ePipeMessageId::pipe_message_beat_heart_id) {
						// beat heart message.
					} else {
						LogAdebug("{} find invalid message id:{}", getPipeIdInfo(this->GetPipeId()), msgId);
						m_state = ePipeState::pipe_session_state_invalid;
						m_session->Close();
					}
				} else if (m_mode == ePipeMode::pipe_mode_connector) {
					// connector peer.
					if (msgId == ePipeMessageId::pipe_message_connect_ack_id) {
						m_state = ePipeState::pipe_session_state_pipe_connected;

						// call pipe connected reporter.
						CPipeModule::instance().GetReporter()->OnReport(true, this);

						// start to send beat heart.
						this->startTimerToSendBeatHeart();
					} else {
						LogAdebug("{} find invalid message id:{}", getPipeIdInfo(GetPipeId()), msgId);
						m_state = ePipeState::pipe_session_state_invalid;
						m_session->Close();
					}
				} else {
					LogAdebug("{} find invalid state:{}", getPipeIdInfo(GetPipeId()), int(m_state));
					m_session->Close();
					return;
				}
			}
			break;

			case ePipeState::pipe_session_state_pipe_connected: {
				if (len == sizeof(anet::tcp::SCommonHead) + sizeof(uint16) + sizeof(heart_beat)) {
					// beat heart message check.
					uint16 msgId = ntohs(*(uint16*)(msg + sizeof(anet::tcp::SCommonHead)));
					if (msgId == ePipeMessageId::pipe_message_beat_heart_id) {
						// excluding pipe message's beat heart id.
						return;
					}
				}

				if (m_pipeHandler != nullptr) {
					// remove the protocol header.
					auto headSize = sizeof(anet::tcp::SCommonHead);
					m_pipeHandler->Handle(msg + headSize, len - headSize);
				} else {
					auto pipeId = getPipeIdInfo(this->GetPipeId());
					LogAdebug("can not find {} pipe's message handler", pipeId);
				}
			}
			break;

			case ePipeState::pipe_session_state_invalid: {
				return;
			}
			break;

			case ePipeState::pipe_session_state_pipe_disconnected: {
				return;
			}
			break;
		  }
		}

		void CPipeSession::startTimerToSendBeatHeart() {
			m_timer.add_repeated_timer([this](void* pData) {
				(void)pData; // remove warning.
				if (m_state == ePipeState::pipe_session_state_pipe_connected) {
					this->sendBeatHeart();
				} else {
					m_timer.kill_timer(uint64(eConst::pipe_beat_heart_send_timer_id));
				}
			}, uint64(eConst::pipe_beat_heart_send_timer_id),
			uint64(eConst::pipe_beat_heart_send_timer_time));
		}

		void CPipeSession::startTimerToCheckBeatHeart() {
			m_lastBeatHeart = GetTickCount();

			// add timer to check beat heart.
			m_timer.add_repeated_timer([this](void* pData) {
				(void)pData; // remove warning.
				if (GetTickCount() - m_lastBeatHeart > gBeatHearTime) {
					LogAdebug("{} has no beat heart now", getPipeIdInfo(m_remoteId));
					m_session->Close();
					m_timer.kill_timer(uint64(eConst::pipe_beat_heart_check_timer_id));
				}
			}, uint64(eConst::pipe_beat_heart_check_timer_id),
			uint64(eConst::pipe_beat_heart_check_timer_time));
		}

		void CPipeSession::sendBeatHeart() {
			heart_beat beatHeart;

			// build heat_beat message.
			std::vector<char> buffer;
			buffer.resize(sizeof(anet::tcp::SCommonHead) + sizeof(uint16) + sizeof(heart_beat));

			char* pBuff = &buffer[0];
			anet::tcp::SCommonHead* pHead = (anet::tcp::SCommonHead*)pBuff;
			pHead->len = decltype(pHead->len)(htonl(sizeof(uint16) + sizeof(heart_beat)));

			// message id
			*(uint16*)(pBuff + sizeof(anet::tcp::SCommonHead)) = htons(uint16(pipe_message_beat_heart_id));
			memcpy(pBuff + sizeof(anet::tcp::SCommonHead) + sizeof(uint16), &beatHeart, sizeof(beatHeart));

			// send to peer.
			m_session->Send(pBuff, int(buffer.size()));
		}

		void CPipeSession::OnConnected() {
			m_state = ePipeState::pipe_session_state_net_connected;
			m_lastBeatHeart = GetTickCount();

			// for connector mode, send connect req.
			if (m_mode == ePipeMode::pipe_mode_connector) {
				this->sendConnectReq();
			} else {
				this->checkStatus();
			}
		}

		void CPipeSession::checkStatus() {
			m_timer.add_once_timer([this](void* pData) {
				if (this->m_state == ePipeState::pipe_session_state_net_connected) {
					LogAdebug("find {} linger so log", this->GetRemotePipeId());
					this->Close();
                }
			}, uint64(eConst::pipe_check_state_timer_id),
			uint64(eConst::pipe_check_state_timer_time));
		}

		void CPipeSession::OnTerminate() {
			// if it is connector, try to reconnect to the remote server.
			if (m_mode == ePipeMode::pipe_mode_connector && 
				m_state == ePipeState::pipe_session_state_pipe_connected) {
				this->reconnect();
			}

			// set pipe disconnected state.
			m_state = ePipeState::pipe_session_state_pipe_disconnected;

			// call OnTerminate().
			CPipeModule::instance().GetReporter()->OnReport(false, this);
		}

		void CPipeSession::reconnect() {
			// try to reconnect.
			auto sessionFactory = CPipeModule::instance().GetSessionFactory();
			auto session = sessionFactory->createSession(this);

			// set new session.
			m_client->setSession(session);
			m_client->reAsyncConnect();
		}

		void CPipeSession::OnRelease() {
			if (m_mode != ePipeMode::pipe_mode_connector) {
				// release me only for pipe_mode_listener.
				// please see the Create() function.
				delete this;
			}
		}

		void CPipeSession::sendConnectAck() {
			connect_ack ack;

			// build connect_ack message.
			std::vector<char> buffer;
			buffer.resize(sizeof(anet::tcp::SCommonHead) + sizeof(uint16) + sizeof(connect_ack));
			char* pBuff = &buffer[0];
			anet::tcp::SCommonHead* pHead = (anet::tcp::SCommonHead*)pBuff;
			pHead->len = (decltype(pHead->len))htonl(sizeof(uint16) + sizeof(connect_ack));

			// message id.
			*(uint16*)(pBuff + sizeof(anet::tcp::SCommonHead)) = htons(uint16(ePipeMessageId::pipe_message_connect_ack_id));
			memcpy(pBuff + sizeof(anet::tcp::SCommonHead) + sizeof(uint16), &ack, sizeof(ack));

			// send to the peer.
			m_session->Send(pBuff, int(buffer.size()));
		}

		void CPipeSession::sendConnectReq() {
			connect_req req;
			{ // init connect_req variable req.
				req.local_id = CPipeModule::instance().GetLocalId();
				req.remote_id = m_remoteId;
				auto minSize = m_token.size() >= sizeof(req.token) ? sizeof(req.token) - 1 : m_token.size();
				strncpy(req.token, m_token.c_str(), minSize);
				req.token[minSize] = 0;
			}
			
			// build connect_req message.
			std::vector<char> buffer;
			buffer.resize(sizeof(anet::tcp::SCommonHead) + sizeof(uint16) + sizeof(connect_req));
			char* pBuff = &buffer[0];
			anet::tcp::SCommonHead* pHead = (anet::tcp::SCommonHead*)pBuff;
			pHead->len = (decltype(pHead->len))htonl(sizeof(uint16) + sizeof(connect_req));

			*(uint16*)(pBuff + sizeof(anet::tcp::SCommonHead)) = htons(uint16(ePipeMessageId::pipe_message_connect_req_id));
			memcpy(pBuff + sizeof(anet::tcp::SCommonHead) + sizeof(uint16), &req, sizeof(req));
			
			// send to peer.
			m_session->Send(pBuff, int(buffer.size()));
		}
	}
}

